昨天的文章中,我們成功建立了一個統一發號施令的 LEDController,並實作了基於 cancel_flag 的搶佔機制。現在,只要呼叫 request_animation(),新的動畫就能立刻中斷並取代舊的動畫。
但這引發了一個嚴重的架構漏洞。
想像 EdgeNode 的運作流程:
request_animation(solid_red) 顯示致命錯誤紅燈。request_animation(breathe_green)。結果:紅燈只亮了三秒,就被不知情的綠燈覆蓋了!
維修人員看到綠燈,以為設備運作正常,殊不知設備早就因為磁碟耗盡而癱瘓。這在醫療儀器或工控領域是絕對不可原諒的。
直覺的做法是在每個發送請求的地方加上狀態判斷:
if current_state != ERROR_STATE:
led_controller.request_animation(breathe_green)
但這會讓整個專案的程式碼變得極度混亂(Spaghetti Code)。狀態機不應該散落在各個業務邏輯中,這違反了我們「解耦」的初衷。
解決方案是為每個任務賦予一個優先權 (Priority)。LEDController 必須記住當前動畫的優先級別,當新任務進來時,只有當新任務的優先級 大於或等於 當前任務時,才允許搶佔。
同時,我們面臨 Python threading 中的一個痛點:我們很難安全且立即地「殺死」一個執行緒。前一篇我們用了一把全域的 _cancel_flag,但如果頻繁切換動畫,這個單一旗標可能會引發時序上的 Race Condition(舊執行緒把新執行緒的旗標給清掉了)。
為此,我們引入資料庫設計常見的概念:世代計數器 (Generation Counter) 或稱 Job ID。
讓我們將 LEDController 升級為最終的企業級版本:
import threading
import time
class LEDController:
# 定義優先權 (數字越大越優先)
PRIO_IDLE = 1
PRIO_SYNCING = 3
PRIO_ERROR = 5
def __init__(self, hardware):
self.hw = hardware
self._current_prio = 0
self._current_job_id = 0
self._lock = threading.Lock()
def _run_anim(self, anim_func, job_id):
# check_cancel 函式:判斷自己這個世代是否已經過期
def check_cancel():
with self._lock:
return self._current_job_id != job_id
# 執行實際動畫
anim_func(self.hw, check_cancel)
def request(self, anim_func, priority):
with self._lock:
# 核心邏輯:低於當前優先權的請求直接被忽略
if priority < self._current_prio:
print(f"[LED] 拒絕低優先級請求 ({priority} < {self._current_prio})")
return False
# 更新狀態與世代交替
self._current_prio = priority
self._current_job_id += 1
new_job_id = self._current_job_id
print(f"[LED] 啟動新動畫 (Prio: {priority}, Job: {new_job_id})")
# 啟動新執行緒
threading.Thread(
target=self._run_anim,
args=(anim_func, new_job_id),
daemon=True
).start()
return True
def clear_error(self):
"""專門用來解除高優先級鎖定的方法"""
with self._lock:
self._current_prio = 0
self.request(breathe_green, self.PRIO_IDLE)
[!TIP]
為何使用 Job ID?
在這種設計中,我們不需要join()等待舊執行緒死掉。只要_current_job_id遞增,舊執行緒在下一次呼叫check_cancel()時,就會發現自己的 job_id != 系統最新的 job_id,然後默默退出。這種孤兒進程自動消亡的模式,極大地提升了系統反應速度。
現在,當磁碟耗盡時,我們發送:led_controller.request(solid_red, LEDController.PRIO_ERROR)
當網路定時檢查完成,它發送:led_controller.request(breathe_green, LEDController.PRIO_IDLE)
LEDController 內部的 self._lock 會發現 PRIO_IDLE (1) < PRIO_ERROR (5),直接退回請求。紅燈得以堅挺地亮著,直到系統管理員前來排除障礙並呼叫 clear_error()。
我們不僅解決了硬體控制的多執行緒問題,還為它建構了一套堅不可摧的狀態機防禦機制。
這就是軟體工程的魅力所在。
明天,我們將探討另一個 Python 開發者的噩夢:time.sleep() 帶來的阻塞與按鍵防彈跳 (Debounce) 的處理策略。Day 23 見!